package chen.bupt.mapreduce.feature;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import chen.bupt.constant.Constants;
import chen.bupt.text.StopWordUtil;
import chen.bupt.text.TextClear;
import chen.bupt.text.TextUtils;
import chen.bupt.util.HDFSFileUtils;

/**
 * 抽取出至少出现过100次的term, 减少特种维度
 * 
 * @author forrest
 * 
 */
public class FeatureExtractJob2 {
	private static String inputPath = Constants.SEG_PATH;
	private static String tmpPath = Constants.TMP2;
	private static String outputPath = "/user/chenlingpeng/smth/feature2";

	public static class Map1 extends
			Mapper<LongWritable, Text, Text, IntWritable> {
		private static final IntWritable one = new IntWritable(1);

		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			if (!HDFSFileUtils.isInputFile(context))
				return;
			String str = value.toString();
			str = TextUtils.splitTextToCont(str, ";", 7)[1];
			if (null == str || "".equals(str))
				return;
			String[] values = str.split(" ");
			for (String feature : values) {
				if ("".equals(feature)) {
					continue;
				}

				if (StopWordUtil.isSpecialTag(feature))
					return;
				if (feature != null) {
					context.write(new Text(feature), one);
				}
			}
		}
	}

	public static class Reduce1 extends
			Reducer<Text, IntWritable, Text, IntWritable> {
		protected void reduce(Text key, Iterable<IntWritable> values,
				Context context) throws IOException, InterruptedException {

			int sum = 0;
			for (IntWritable value : values) {
				sum += value.get();
			}
			String keyS = key.toString();
			
			if(StopWordUtil.isSpecialTag(keyS)){
				return;
			}
			context.write(new Text(keyS), new IntWritable(sum));
		}
	}

	public static class Map2 extends Mapper<LongWritable, Text, Text, Text> {
		private static final Text one = new Text("key");

		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			if (!HDFSFileUtils.isInputFile(context))
				return;
			context.write(one, value);
		}
	}

	public static class Reduce2 extends Reducer<Text, Text, Text, NullWritable> {
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			StringBuffer sb = new StringBuffer();
			int count = 0;
			for (Text val : values) {
				String[] kv = val.toString().split("\t");
				if(kv[0].contains("/nr")){
					if(Integer.parseInt(kv[1])<10){
						continue;
					}else{
						kv[0] = TextClear.substring(kv[0], "/nr");
					}
				}
				count++;
				sb.append(kv[0] + ":" + kv[1] + ";");
			}
			context.write(new Text(sb.toString()), NullWritable.get());
			context.write(new Text("total feature: " + count), NullWritable
					.get());
		}
	}

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		Job job = new Job(conf, "featureExtract");
		job.setJarByClass(FeatureExtractJob2.class);
		job.setMapperClass(Map1.class);
		job.setCombinerClass(Reduce1.class);
		job.setReducerClass(Reduce1.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		Path input = new Path(inputPath);
		Path tmp = new Path(tmpPath);
		Path output = new Path(outputPath);
		HDFSFileUtils.deleteFile(output, conf);
		HDFSFileUtils.deleteFile(tmp, conf);
		FileInputFormat.addInputPath(job, input);
		FileOutputFormat.setOutputPath(job, tmp);
		job.waitForCompletion(true);

		Configuration conf2 = new Configuration();
		Job job2 = new Job(conf2, "featureExtract2");
		job2.setJarByClass(FeatureExtractJob2.class);
		job2.setMapperClass(Map2.class);
		job2.setReducerClass(Reduce2.class);
		job2.setOutputKeyClass(Text.class);
		job2.setOutputValueClass(Text.class);
		FileInputFormat.addInputPath(job2, tmp);
		FileOutputFormat.setOutputPath(job2, output);
		job2.waitForCompletion(true);

	}

}
